/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.reduce.klsh.kmeaning.mappers;

import aiaudio.Application;
import aiaudio.processing.reduce.klsh.kmeaning.DictionaryCoordinatePair;
import aiaudio.database.tables.RatingsMatrixTable;
import aiaudio.database.tables.ReduceDictionaryTable;
import java.io.IOException;
import java.util.NavigableMap;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author nastya
 */
public class KMeansAlgorithmMapper extends TableMapper<ImmutableBytesWritable, DictionaryCoordinatePair> {

    public static final String DICTIONARY_TABLE_NAME = "dictionaryTableName";
    public static final String RATINGS_SOURCE_NAME = "ratingsSourceName";
    private HTable dictionaryTable;
    private HTable ratingsSourceTable;
    private Result defaultValuesByUsers;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        setupDictionary(context);
        setupRatingsResource(context);
        setupDefaultRatingsByUser();
    }

    protected void setupDefaultRatingsByUser() throws IOException {
        defaultValuesByUsers = RatingsMatrixTable.getDefaultRatings(ratingsSourceTable);
    }

    protected void setupRatingsResource(Context context) {
        String ratingsSourceName = context.getConfiguration().get(RATINGS_SOURCE_NAME);
        ratingsSourceTable = Application.database().getByName(ratingsSourceName).getTable();
    }

    protected void setupDictionary(Context context) {
        String dictionaryTableName = context.getConfiguration().get(DICTIONARY_TABLE_NAME);
        dictionaryTable = Application.database().getByName(dictionaryTableName).getTable();
    }

    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        Scan scan = new Scan();
        ResultScanner scanner = dictionaryTable.getScanner(scan);
        Result dictResult = null;
        while ((dictResult = scanner.next()) != null) {
            double minDistance = Double.MAX_VALUE;
            byte[] minCoordinateNum = Bytes.toBytes("0");
                    
            byte[] dictKey = dictResult.getRow();
            NavigableMap<byte[], byte[]> allCoordinates = ReduceDictionaryTable.getCentroids(dictResult);            
            for (byte[] coordinateNum : allCoordinates.keySet()) {
                byte[] coordinateValue = allCoordinates.get(coordinateNum);

                Get get = new Get(coordinateValue);
                Result coordinateResult = ratingsSourceTable.get(get);
                double distance = calculateDistance(value, coordinateResult, defaultValuesByUsers);
                if(distance < minDistance){
                    minDistance = distance;
                    minCoordinateNum = coordinateNum;
                }
            }
            DictionaryCoordinatePair dictionaryCoordinatePair = new DictionaryCoordinatePair(Bytes.toString(dictKey), Bytes.toString(minCoordinateNum), minDistance);
            context.write(new ImmutableBytesWritable(key), dictionaryCoordinatePair);
        }

    }

    private double calculateDistance(Result value, Result coordinateResult, Result defaultValuesByUsers) {
        NavigableMap<byte[], byte[]> allUsers = RatingsMatrixTable.getUsers(defaultValuesByUsers);
        double sum = 0;
        for (byte[] user : allUsers.keySet()) {
            double val1 = getRatingOrDafaultByUser(value, user, defaultValuesByUsers);
            double val2 = getRatingOrDafaultByUser(coordinateResult, user, defaultValuesByUsers);
            sum += (val1 - val2) * (val1 - val2);
        }
        return Math.sqrt(sum);
    }

    private double getRatingOrDafaultByUser(Result rMain, byte[] user, Result rDefault) {
        return RatingsMatrixTable.getArtistRatingOrDefault(rMain, rDefault, user);
    }
}
